package net.sf.calibur.client;

import java.io.ByteArrayOutputStream;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.net.SocketAddress;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.LinkedBlockingQueue;

import net.sf.calibur.msgbuf.Command;
import net.sf.calibur.msgbuf.CommandFactory;
import net.sf.calibur.transport.message.CommandMessage;
import net.sf.calibur.transport.message.HeartBeatMessage;
import net.sf.calibur.transport.message.Message;
import net.sf.calibur.transport.message.RequestSessionMessage;
import net.sf.calibur.transport.message.SessionEstablishedMessage;

class Sender implements Runnable {
	SocketClientSession session;
	private Logger logger;
	private DataOutputStream netDos;
	private ByteArrayOutputStream baos = new ByteArrayOutputStream();
	private LinkedBlockingQueue<Message> queue = new LinkedBlockingQueue<Message>();

	public Sender(SocketClientSession session, Logger logger) {
		this.session = session;
		this.logger = logger;
	}

	public void reset(DataOutputStream dos) {
		this.netDos = dos;
		this.baos.reset();
		this.queue.clear();
	}

	public void send(Message message) {
		queue.add(message);
	}

	public void sendMessage(Message message) throws Exception {
		baos.reset();
		DataOutputStream dos = new DataOutputStream(baos);
		
		byte messageType = (byte) (message.getType() & 0x0f);
		if (message.getType() == Message.MT_APPLICATION) {
			messageType |= 0x10;
		}
		dos.writeByte(messageType);
		
		if (message.getType() == Message.MT_APPLICATION) {
			dos.writeLong(System.currentTimeMillis());
		}

		if (message.getType() == Message.MT_APPLICATION) {
			CommandMessage cm = (CommandMessage) message;
			dos.writeShort(cm.getCommandId());
			cm.getCommand().serializeInputArgument(baos);
		} else {
			message.serialize(baos);
		}

		int msgSize = baos.size();
		netDos.writeInt(msgSize);
		baos.writeTo(netDos);
	}

	@Override
	public void run() {
		Message message = null;

		while (true) {
			try {
				message = queue.take();
			} catch (Exception e) {
				logger.warn("Sender take message failed.", e);
				session.setError(e);
				return;
			}

			try {
				sendMessage(message);
			} catch (Exception e) {
				logger.error("Sender send message failed.", e);
				session.setError(e);
				return;
			}
		}
	}
}

class Receiver implements Runnable {
	SocketClientSession session;
	private Logger logger;
	private DataInputStream netDis;
	private ConcurrentLinkedQueue<Message> queue = new ConcurrentLinkedQueue<Message>();
	private Message heartbeatAck = new HeartBeatMessage(
			Message.MT_SERVER_HEARTBEAT_ACK);

	public Receiver(SocketClientSession session, Logger logger) {
		this.session = session;
		this.logger = logger;
	}

	public void reset(DataInputStream dis) {
		this.netDis = dis;
		this.queue.clear();
	}

	public Message getMessage() {
		return queue.poll();
	}

	@Override
	public void run() {
		Message message = null;

		while (true) {
			try {
				message = recvMessage();
				switch (message.getType()) {
				case Message.MT_SERVER_HEARTBEAT:
					session.send(heartbeatAck);
					break;

				case Message.MT_CLIENT_HEARTBEAT_ACK:
					break;

				case Message.MT_APPLICATION:
					queue.add(message);
					break;

				case Message.MT_APPLICATION_ACK:
					break;
					
				default:
					throw new IllegalArgumentException(
							"Unsupported message type " + message.getType());
				}
			} catch (Exception e) {
				logger.error("Receiver recv message failed.", e);
				session.setError(e);
				return;
			}
		}
	}

	public Message recvMessage() throws Exception {
		int msgSize = netDis.readInt();
		byte msgType = (byte) (netDis.readByte() & 0x0f);
		Message message = Message.createMessage(msgType,
				session.getCommandFactory());

		if (msgType == Message.MT_APPLICATION) {
			int commandId = netDis.readShort();
			CommandMessage cm = (CommandMessage) message;
			Command command = cm.createCommand(commandId);
			command.deserializeOutputArgument(netDis);
		} else {
			message.deserialize(netDis);
		}

		logger.debug("Received message size=" + msgSize + " type=" + msgType);
		return message;
	}
}

public class SocketClientSession implements ClientSession {
	private CommandFactory commandFactory;
	private SocketAddress serverAddress;
	private Socket socket;
	private String key;
	private long establishedTimestamp;
	private DataInputStream netDis;
	private DataOutputStream netDos;
	private Throwable error;
	private Logger logger;
	private Sender sender;
	private Receiver receiver;
	private Thread sendThread, recvThread;

	public SocketClientSession(CommandFactory commandFactory) {
		this(commandFactory, new ConsoleLogger());
	}

	public SocketClientSession(CommandFactory commandFactory, Logger logger) {
		this.logger = logger;
		this.commandFactory = commandFactory;
		sender = new Sender(this, logger);
		receiver = new Receiver(this, logger);
	}

	public CommandFactory getCommandFactory() {
		return commandFactory;
	}

	@Override
	public void connect(String host, int port) throws Exception {
		serverAddress = new InetSocketAddress(host, port);

		// 与服务器建立TCP连接
		createTcpLink();

		// 与服务器建立session
		createSession();

		// 启动发送/接收线程
		start();
	}

	@Override
	public String getKey() {
		return key;
	}

	@Override
	public long getEstablishedTimestamp() {
		return establishedTimestamp;
	}

	public void setError(Throwable t) {
		this.error = t;
	}

	private void start() {
		sender.reset(netDos);
		receiver.reset(netDis);

		sendThread = new Thread(sender);
		recvThread = new Thread(receiver);

		sendThread.start();
		recvThread.start();
	}

	private void createTcpLink() throws Exception {
		logger.debug("Creating TCP link with server address " + serverAddress
				+ " ...");

		socket = new Socket();
		socket.connect(serverAddress);
		netDis = new DataInputStream(socket.getInputStream());
		netDos = new DataOutputStream(socket.getOutputStream());

		sender.reset(netDos);
		receiver.reset(netDis);
	}

	private void createSession() throws Exception {
		logger.debug("Creating application level session with server ...");

		RequestSessionMessage rsm = new RequestSessionMessage();
		sender.sendMessage(rsm);

		SessionEstablishedMessage ses = (SessionEstablishedMessage) receiver
				.recvMessage();
		this.key = ses.getSessionKey();
		this.establishedTimestamp = ses.getEstablishedTimestamp();

		logger.debug("Session created with key=" + key);
	}

	@Override
	public synchronized void close() {
		if (socket != null) {
			try {
				socket.close();
			} catch (Exception e) {
				logger.warn("Close socket fail.", e);
			}

			if (sendThread != null) {
				try {
					sendThread.interrupt();
					sendThread.join(5000L);
				} catch (Exception e) {
					logger.warn("Interrupt send thread failed.", e);
				}
			}

			if (recvThread != null) {
				try {
					recvThread.interrupt();
					recvThread.join(5000L);
				} catch (Exception e) {
					logger.warn("Interrupt recv thread failed.", e);
				}
			}

			socket = null;
			sendThread = null;
			recvThread = null;
			netDis = null;
			netDos = null;

			serverAddress = null;
			key = null;
			establishedTimestamp = -1;
			error = null;
		}
	}

	protected void send(Message message) {
		sender.send(message);
	}

	@Override
	public void sendCommand(Command command) throws Exception {
		if (error != null) {
			throw new Exception("Error session.", error);
		}

		Message message = new CommandMessage(commandFactory, command);
		send(message);
	}

	@Override
	public Command recvCommand() throws Exception {
		if (error != null) {
			throw new Exception("Error session.", error);
		}

		CommandMessage cm = (CommandMessage) receiver.getMessage();
		return cm == null ? null : cm.getCommand();
	}
}
